package com.zjl.Kafka.第03章_操作kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.HashMap;
import java.util.Map;

/**
 * 消费者组   必选项
 *      一个消费者组当成了一个整体
 *
 *  1个分区，只能被一个消费者消费(除非此消费者挂了)，但是一个消费者可以消费多个分区
 *
 *  分区：消费者    n:1
 *
 *  比如 五个分区，但是这个消费者组有6个消费者，就会有一个消费者啥也不干，
 *          这个消费者就是备用,当有消费者挂了，就会替代
 * *****************************************************************************
 * 消费者组偏移量文件  __consumer_offsets-xx  xx -> 0-49
 *    消费者组名字.hashcode % 50
 *
 *
 * 消费者分配策略 Assignor
 *      这个分配策略其实是由消费者的 主消费者（Leader）决定的，这个Leader我们称之为群主。
 *      群主是多个消费者中，第一个加入组中的消费者，其他消费者我们称之为Follower,
 *      称呼上有点类似与分区的Leader和Follower.
 *
 *
 *      当消费者加入群组的时候，会发送一个JoinGroup请求。
 *      群主负责给每一个消费者分配分区。
 *
 *  Kafka提供的分区分配策略常用的有4个:
 *   每个消费者组中的消费者都会含有一个自动生产的UUID作为 memberid。
 *      ➢ RoundRobinAssignor (轮询分配策略)    会导致靠前的 memberid 消费的分区多
 *              轮询策略中会将每个消费者按照memberid进行排序，
 *              所有member消费的主题分区根据主题名称进行排序。
 *      ➢RangeAssignor (范围分配策略)         会导致靠前的 memberid 消费的分区多
 *              按照每个topic 的partition数计算出每个消费者应该分配的分区数量，然后分配，
 *              分配的原则就是一个主题的分区尽可能的平均分，
 *              如果不能平均分，那就按顺序向前补齐即可。
 *      ➢StickyAssignor (粘性分区)
 *              在第一次分配后，每个组成员都保留分配给自己的分区信息。
 *              如果有消费者加入或退出，那么在进行分区再分配时
 *              (一般情况下，消费者退出45s后，才会进行再分配，因为需要考虑可能又恢复的情况)，
 *              尽可能保证消费者原有的分区不变，重新对加入或退出消费者的分区进行分配。
 *      ➢CooperativeStickyAssignor
 *              前面的三种分配策略再进行重分配时使用的是EAGER协议，
 *              会让当前的所有消费者放弃当前分区，关闭连接，资源清理，重新加入组和等待分配策略。
 *              明显效率是比较低的，所以从Kafka2.4版本开始，在粘性分配策略的基础上，优化了重分配的过程,
 *              使用的是COOPERATIVE协议，特点就是在整个再分配的过程中可以看出，
 *              粘性分区分配策略分配的会更加均匀和高效一些 ，
 *              COOPERATIVE协议将一次全局重平衡，改成每次小规模重平衡，直至最终收敛平衡的过程。
 *   Kafka消费者默认的分区分配就是 RangeAssignor(第一次), CooperativeStickyAssignor(后续)
 *
 *
 */
public class E3_消费者组的消费策略 {
    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "82.157.71.243:9092");//连接地址
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//key 的 反序列化
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());//value 的 反序列化

        config.put(ConsumerConfig.GROUP_ID_CONFIG, "x01");//消费者组
        config.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG,"001");//每个组的 消费者id
        config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, ConsumerPartitionAssignor.Assignment.class.getName());

    }

}
